﻿// Copyright (c) Inossem. All rights reserved.
// Licensed under the LGPL-3.0 license. See LICENSE file in the project root for full license information.
using NamedPipeWrapper.Common;
using NamedPipeWrapper.IO;
using NamedPipeWrapper.Threading;
using NLog;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.IO.Pipes;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace NamedPipeWrapper
{
    public class NamedPipeServerExAsync<TRead, TWrite>
        where TRead : class
        where TWrite : class
    {
        /// <summary>
        /// Invoked whenever a client connects to the server.
        /// </summary>
        public event ConnectionExAsyncEventHandler<TRead, TWrite> ClientConnected;

        /// <summary>
        /// Invoked whenever a client disconnects from the server.
        /// </summary>
        public event ConnectionExAsyncEventHandler<TRead, TWrite> ClientDisconnected;

        /// <summary>
        /// Invoked whenever a client sends a message to the server.
        /// </summary>
        public event ConnectionExAsyncMessageEventHandler<TRead, TWrite> ClientMessage;

        /// <summary>
        /// Invoked whenever an exception is thrown during a read or write operation.
        /// </summary>
        public event PipeExceptionEventHandler Error;

        private readonly string _pipeName;
        private readonly PipeSecurity _pipeSecurity;
        private readonly ConcurrentDictionary<string, NamedPipeConnectionExAsync<TRead, TWrite>> _connections = new ConcurrentDictionary<string, NamedPipeConnectionExAsync<TRead, TWrite>>();

        private int _nextPipeId;
        private volatile bool _shouldKeepRunning;

        public NamedPipeServerExAsync(string pipeName)
          : this(pipeName, null)
        {
        }

        /// <summary>
        /// Constructs a new <c>NamedPipeServer</c> object that listens for client connections on the given <paramref name="pipeName"/>.
        /// </summary>
        /// <param name="pipeName">Name of the pipe to listen on</param>
        public NamedPipeServerExAsync(string pipeName, PipeSecurity pipeSecurity)
        {
            _pipeName = pipeName;
            _pipeSecurity = pipeSecurity;
        }

        /// <summary>
        /// Begins listening for client connections in a separate background thread.
        /// This method returns immediately.
        /// </summary>
        public void Start()
        {
            _shouldKeepRunning = true;
            Task.Run(listen);
        }

        public void PushMessage(TWrite message)
        {
            foreach (var pair in _connections)
            {
                var client = pair.Value;
                if (client.IsConnected)
                {
                    client.PushMessage(message);
                }
            }
        }

        public void PushMessage(TWrite message, string clientName)
        {
            NamedPipeConnectionExAsync<TRead, TWrite> client = null;
            var isok = _connections.TryGetValue(clientName, out client);

            if (isok && client.IsConnected)
            {
                client.PushMessage(message);
            }
        }

        /// <summary>
        /// Sends a message to all connected clients asynchronously.
        /// This method returns immediately, possibly before the message has been sent to all clients.
        /// </summary>
        /// <param name="message"></param>
        public async Task PushMessageAsync(TWrite message)
        {
            foreach (var pair in _connections)
            {
                var client = pair.Value;
                if (client.IsConnected)
                {
                    await pair.Value.PushMessageAsync(message);
                }
            }
        }

        /// <summary>
        /// push message to the given client.
        /// </summary>
        /// <param name="message"></param>
        /// <param name="clientName"></param>
        public async Task PushMessageAsync(TWrite message, string clientName)
        {
            NamedPipeConnectionExAsync<TRead, TWrite> client = null;
            var isok = _connections.TryGetValue(clientName, out client);

            if (isok && client.IsConnected)
            {
                await client.PushMessageAsync(message);
            }
        }

        /// <summary>
        /// Closes all open client connections and stops listening for new ones.
        /// </summary>
        public void Stop()
        {
            _shouldKeepRunning = false;

            foreach (var pair in _connections)
            {
                pair.Value.Close();
            }
        }

        #region Private methods

        private void listen()
        {
            try
            {
                while (_shouldKeepRunning)
                {
                    waitForConnection(_pipeName, _pipeSecurity);
                }
            }
            catch (Exception ex)
            {
                LogHelper.Error(ex);
                OnError(ex);
            }
        }

        private void waitForConnection(string pipeName, PipeSecurity pipeSecurity)
        {
            NamedPipeServerStream handshakePipe = null;
            var autoResetEvent = new AutoResetEvent(false);
            var connectionPipeName = getNextConnectionPipeName(pipeName);

            try
            {
                // Send the client the name of the data pipe to use
                handshakePipe = PipeServerExFactory.CreateAndConnectPipe(pipeName, pipeSecurity);

                Task.Run(() =>
                {
                    NamedPipeServerStream dataPipe = null;
                    NamedPipeConnectionExAsync<TRead, TWrite> connection = null;
                    try
                    {
                        // Wait for the client to connect to the data pipe
                        dataPipe = PipeServerExFactory.CreatePipe(connectionPipeName, pipeSecurity);
                        autoResetEvent.Set();
                        if (dataPipe.WaitForConnectionAsync().Wait(TimeSpan.FromSeconds(10)))
                        {
                            // Add the client's connection to the list of connections
                            connection = ConnectionExAsyncFactory.CreateConnection<TRead, TWrite>(dataPipe);
                            connection.ReceiveMessage += ClientOnReceiveMessage;
                            connection.Disconnected += ClientOnDisconnected;
                            connection.Error += ConnectionOnError;
                            connection.Open();

                            _connections.TryAdd(connection.Name, connection);

                            ClientOnConnected(connection);
                        }
                        else
                        {
                            dataPipe.Close();
                        }
                    }
                    catch (Exception e)
                    {
                        LogHelper.Error(e);
                        cleanup(dataPipe);
                        ClientOnDisconnected(connection);
                    }
                });

                autoResetEvent.WaitOne();
                var handshakeWrapper = new PipeStreamWrapperEx<string, string>(handshakePipe);
                handshakeWrapper.WriteObject(connectionPipeName);
                handshakeWrapper.Close();
            }
            catch (Exception e)
            {
                LogHelper.Error(e);
                cleanup(handshakePipe);
            }
            finally
            {
                autoResetEvent.Dispose();
            }
        }

        protected void ClientOnConnected(NamedPipeConnectionExAsync<TRead, TWrite> connection)
        {
            if (ClientConnected != null)
                ClientConnected(connection);
        }

        protected void ClientOnReceiveMessage(NamedPipeConnectionExAsync<TRead, TWrite> connection, TRead message)
        {
            if (ClientMessage != null)
                ClientMessage(connection, message);
        }

        protected void ClientOnDisconnected(NamedPipeConnectionExAsync<TRead, TWrite> connection)
        {
            if (connection == null)
                return;

            NamedPipeConnectionExAsync<TRead, TWrite> temp;
            _connections.TryRemove(connection.Name, out temp);

            if (ClientDisconnected != null)
                ClientDisconnected(connection);
        }

        /// <summary>
        ///     Invoked on the UI thread.
        /// </summary>
        protected void ConnectionOnError(NamedPipeConnectionExAsync<TRead, TWrite> connection, Exception exception)
        {
            OnError(exception);
        }

        /// <summary>
        ///     Invoked on the UI thread.
        /// </summary>
        /// <param name="exception"></param>
        protected void OnError(Exception exception)
        {
            if (Error != null)
                Error(exception);
        }

        private string getNextConnectionPipeName(string pipeName)
        {
            return string.Format("{0}_{1}", pipeName, ++_nextPipeId);
        }

        private static void cleanup(NamedPipeServerStream pipe)
        {
            if (pipe == null) return;
            using (var x = pipe)
            {
                x.Close();
            }
        }

        #endregion
    }

    static class PipeServerExFactory
    {
        public static NamedPipeServerStream CreateAndConnectPipe(string pipeName, PipeSecurity pipeSecurity)
        {
            var pipe = CreatePipe(pipeName, pipeSecurity);
            pipe.ReadMode = PipeTransmissionMode.Message;
            pipe.WaitForConnection();
            return pipe;
        }

        public static NamedPipeServerStream CreatePipe(string pipeName, PipeSecurity pipeSecurity)
        {
            return new NamedPipeServerStream(pipeName, PipeDirection.InOut, 1, PipeTransmissionMode.Message, PipeOptions.Asynchronous | PipeOptions.WriteThrough, 0, 0, pipeSecurity);
        }
    }
}
